package net.bwie.realtime.dwd.douyin.logs.job;


import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import net.bwie.realtime.dwd.douyin.logs.bean.EventLog;
import net.bwie.realtime.dwd.douyin.logs.bean.TradePay;
import net.bwie.realtime.dwd.douyin.logs.bean.TradePayBeanFull;
import net.bwie.realtime.jtp.utils.DateTimeUtil;
import net.bwie.realtime.jtp.utils.KafkaUtil;
import net.bwie.realtime.jtp.utils.KafkaUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

import java.time.Duration;

public class TransactionAmountJob {
    public static void main(String[] args) throws Exception{
    //1.执行环境
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
//        env.enableCheckpointing(3000L);
    //2.数据源
    DataStream<String> dwd_event_log = KafkaUtil.consumerKafka(env, "dwd_event_log");
    DataStream<String> dwd_trade_pay_detail = KafkaUtil.consumerKafka(env, "dwd_trade_pay");
//    dwd_trade_pay_detail.print();


    DataStream<String> handleStream = handle(dwd_event_log, dwd_trade_pay_detail);
    handleStream.print();
//    // 执行任务
    env.execute("TransactionAmountJob");
}

    private static DataStream<String> handle(DataStream<String> dwd_event_log, DataStream<String> dwd_trade_pay_detail) {
        // 转实体类
        SingleOutputStreamOperator<EventLog> EventLogStream = dwd_event_log.map(new MapFunction<String, EventLog>() {
            @Override
            public EventLog map(String s) throws Exception {
                JSONObject jsonObject = JSON.parseObject(s);
                String leave_time = jsonObject.getString("leave_time");
                int live_room_id = jsonObject.getIntValue("live_room_id");
                Integer anchor_id = jsonObject.getInteger("anchor_id");
                String session_id = jsonObject.getString("session_id");
                String sku_id = jsonObject.getString("sku_id");
                String source_type = jsonObject.getString("source_type");
                String comment_id = jsonObject.getString("comment_id");
                String enter_time = jsonObject.getString("enter_time");
                int duration_sec = jsonObject.getIntValue("duration_sec");
                String platform = jsonObject.getString("platform");
                String event_id = jsonObject.getString("event_id");
                String event_type = jsonObject.getString("event_type");
                String user_id = jsonObject.getString("user_id");
                String live_session_id = jsonObject.getString("live_session_id");
                String sku_name = jsonObject.getString("sku_name");
                String category = jsonObject.getString("category");
                String region = jsonObject.getString("region");
                return new EventLog(
                        leave_time,
                        live_room_id,
                        anchor_id,
                        session_id,
                        sku_id,
                        source_type,
                        comment_id,
                        enter_time,
                        duration_sec,
                        platform,
                        event_id,
                        event_type,
                        user_id,
                        live_session_id,
                        sku_name,
                        category,
                        region
                );
            }
        });
        SingleOutputStreamOperator<EventLog> eventLogTime = EventLogStream.assignTimestampsAndWatermarks(
                WatermarkStrategy.<EventLog>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                        .withTimestampAssigner(new SerializableTimestampAssigner<EventLog>() {
                            @Override
                            public long extractTimestamp(EventLog eventLog, long l) {
                                //事件时间 处理乱序迟到数据
                                Long ts = DateTimeUtil.convertStringToLong(eventLog.getEnterTime(), "yyyy-MM-dd HH:mm:ss");
                                return ts;
                            }
                        })
        );
        SingleOutputStreamOperator<TradePay> TradePayStream = dwd_trade_pay_detail.map(new MapFunction<String, TradePay>() {
            @Override
            public TradePay map(String s) throws Exception {
                JSONObject jsonObject = JSON.parseObject(s);
                String pay_status = jsonObject.getString("pay_status");
                Integer anchor_id = jsonObject.getInteger("anchor_id");
                String user_id = jsonObject.getString("user_id");
                Double pay_amount = jsonObject.getDoubleValue("pay_amount");
                String live_session_id = jsonObject.getString("live_session_id");
                String session_id = jsonObject.getString("session_id");
                String sku_id = jsonObject.getString("sku_id");
                String sku_name = jsonObject.getString("sku_name");
                String pay_id = jsonObject.getString("pay_id");
                String category = jsonObject.getString("category");
                String order_id = jsonObject.getString("order_id");
                String pay_time = jsonObject.getString("pay_time");
                return new TradePay(
                        pay_status,
                        anchor_id,
                        user_id,
                        pay_amount,
                        live_session_id,
                        session_id,
                        sku_id,
                        sku_name,
                        pay_id,
                        category,
                        order_id,
                        pay_time
                );
            }
        });
        SingleOutputStreamOperator<TradePay> tradePayTime = TradePayStream.assignTimestampsAndWatermarks(
                WatermarkStrategy.<TradePay>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                        .withTimestampAssigner(new SerializableTimestampAssigner<TradePay>() {
                            @Override
                            public long extractTimestamp(TradePay tradePay, long l) {
                                Long ts = DateTimeUtil.convertStringToLong(tradePay.getPayTime(), "yyyy-MM-dd HH:mm:ss");
                                return ts;
                            }
                        })
        );
//        //  分组
        KeyedStream<EventLog, Integer> eventLogKey = eventLogTime.keyBy(bean -> bean.getAnchorId());
        KeyedStream<TradePay, Integer> tradePayKey = tradePayTime.keyBy(bean -> bean.getAnchorId());

        // 定义侧输出标签用于处理未匹配的数据
//        OutputTag<EventLog> unmatchedEventTag = new OutputTag<EventLog>("unmatched-events") {};
//        OutputTag<TradePay> unmatchedTradeTag = new OutputTag<TradePay>("unmatched-trades") {};
//        SingleOutputStreamOperator<String> process = eventLogKey
//                .intervalJoin(tradePayKey)
//                .between(Time.seconds(-5), Time.seconds(2))
//                .sideOutputLeftLateData(unmatchedEventTag)
//                .sideOutputRightLateData(unmatchedTradeTag)
//                .process(new ProcessJoinFunction<EventLog, TradePay, String>() {
//                    @Override
//                    public void processElement(EventLog eventLog, TradePay tradePay, Context context, Collector<String> collector) throws Exception {
//                        Integer liveRoomId = eventLog.getLiveRoomId();
//                        String userId = eventLog.getUserId();
//                        String enterTime = eventLog.getEnterTime();
//                        String orderId = tradePay.getOrderId();
//                        String payTime = tradePay.getPayTime();
//
//                        String result = liveRoomId + "," + userId + "," + enterTime + "," + orderId + "," + payTime;
//                        collector.collect(result);
//                    }
//                });

        SingleOutputStreamOperator<TradePayBeanFull> tradePayFull = eventLogKey
                .intervalJoin(tradePayKey)
                .between(Time.seconds(-5), Time.seconds(10))
                .process(new ProcessJoinFunction<EventLog, TradePay, TradePayBeanFull>() {
                    @Override
                    public void processElement(EventLog eventLog, TradePay tradePay, ProcessJoinFunction<EventLog, TradePay, TradePayBeanFull>.Context context, Collector<TradePayBeanFull> collector) throws Exception {
                        String payStatus = tradePay.getPayStatus();
                        Integer anchorId = tradePay.getAnchorId();
                        String userId = tradePay.getUserId();
                        String payId = tradePay.getPayId();
                        String orderId = tradePay.getOrderId();
                        String payTime = tradePay.getPayTime();
                        Integer liveRoomId = eventLog.getLiveRoomId();
                        collector.collect(new TradePayBeanFull(
                                payStatus,
                                anchorId,
                                userId,
                                payId,
                                orderId,
                                payTime,
                                liveRoomId
                        ));
                    }
                });
        SingleOutputStreamOperator<String> processStream = tradePayFull.process(
                new ProcessFunction<TradePayBeanFull, String>() {
            @Override
            public void processElement(TradePayBeanFull s,
                                       Context context, Collector<String> collector) throws Exception {
                String payStatus = s.getPayStatus();
                Integer anchorId = s.getAnchorId();
                String userId = s.getUserId();
                String payId = s.getPayId();
                String orderId = s.getOrderId();
                String payTime = s.getPayTime();
                Integer liveRoomId = s.getLiveRoomId();
                String output = payStatus + "," + anchorId + "," + userId +
                        "," + payId + "," + orderId + "," + payTime + "," + liveRoomId;
                collector.collect(JSON.toJSONString(output));
            }
        });

//        KafkaUtils.producerKafkaObject(processStream,"dws_event_pay_transaction");
        return processStream;
    }
}
